﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System.Threading.Actors;
using System;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Diagnostics;
using System.Diagnostics.Contracts;
using System.IO;
using System.Linq;
using System.Reactive;
using System.Reflection;
using System.Runtime.InteropServices;
using System.Runtime.Serialization;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Runtime.CompilerServices;

namespace System.Cloud.Collections
{
    public class CloudList<T> : IListAsync<T>, IList<T>, IDisposable
    {
        private ActorClient _actorClient;
        private String _listName;
        private bool _deleteActor;
        private const String ApplicationType = "fabric:/actor/list";

        private readonly Object[] ListNameOneElementArray;

        private string _partitionName;

        // @TODO: Think of other constructor parameters
        // Persistence mode - InMemory vs. PersistedToDisk
        // Responsiveness - optimize for latency or throughput?  IE, our actor runtime's 100 ms BatchAcknowledgementInterval
        // Should we cache anything locally?  If so, how frequently do we re-poll?  Or should users use ObservableCloudList instead?
        public CloudList(String server, int port, String listName)
            : this(server + ":" + port, listName, /*CancellationToken.None,*/ false)
        {
            Contract.Requires(!String.IsNullOrEmpty(listName));
        }

        /*
        public CloudList(String server, int port, String listName, CancellationToken cancellationToken)
            : this(server + ":" + port, listName, cancellationToken)
        {
        }
        */
        public CloudList(String hostEndpoint, String listName)
            : this(hostEndpoint, listName, /*CancellationToken.None,*/ false)
        {
        }

        /*
        public CloudList(String hostEndpoint, String listName, CancellationToken cancellationToken)
            : this(hostEndpoint, listName, cancellationToken, false)
        {
        }
        */

        public CloudList(String hostEndpoint, String listName, /*CancellationToken cancellationToken,*/ bool connectThroughGateway, int numReplicas = 2)
        {
            Contract.Requires(!String.IsNullOrEmpty(hostEndpoint));
            Contract.Requires(!String.IsNullOrEmpty(listName));
            Contract.Requires(numReplicas >= 1, "Actors must have at least one instance of their state in the cloud.");
            
            Uri actorUri = new Uri(ApplicationType + "/" + listName);
            _listName = listName;
            ListNameOneElementArray = new Object[] { _listName };

            // See if this collection exists.  If not, create it.
            _actorClient = ActorClient.CreateOrConnectToActor(hostEndpoint, ApplicationType, actorUri, 
                                                              "ListActorReplica", connectThroughGateway, numReplicas);

            // Deploy T's assembly to the server, if necessary.
            // @TODO: Look at types T depends on.  IE, is T a Tuple<UserType1, UserType2>?
            // Update: We no longer need .NET type information on the actor.  It's all JSON on the actor.
            //_actorClient.DeployAssemblyIfNeeded(typeof(T).GetTypeInfo().Assembly);

            _actorClient.CallMethod("TryConstruct", ListNameOneElementArray);
            //_cancellationToken = cancellationToken;
        }

#if CUSTOM_PARTITIONING_AND_CONNECTION_SHARING
        // Manual partitioning approach, coupled with connection pooling for performance.
        // @TODO: If we enable this, visit Dispose and ensure we don't dispose pooled actor clients.
        public CloudList(String hostEndpoint, String listName, Func<String, String> partitioner, /*CancellationToken cancellationToken,*/ bool connectThroughGateway, int numReplicas = 2)
        {
            Contract.Requires(!String.IsNullOrEmpty(hostEndpoint));
            Contract.Requires(!String.IsNullOrEmpty(listName));
            Contract.Requires(partitioner != null);
            Contract.Requires(numReplicas >= 1, "Actors must have at least one instance of their state in the cloud.");

            _partitionName = partitioner(listName);
            Uri actorUri = new Uri(ApplicationType + "/" + _partitionName);
            _listName = listName;
            ListNameOneElementArray = new Object[] { listName };

            // This solution required a ConditionalWeakTable or a Dictionary, mapping partition names to FabricActorClient
            // instances.  ConditionalWeakTable is much slower than Dictionary, but with Dictionary we'd need a ref 
            // counting scheme (or rely on finalization)!  However a VS perf expert pointed out that ConditionalWeakTable
            // takes a lock in TryGetValue, and I'm not doing that with Dictionary.

            // Try finding an already-existing FabricActorClient to the right partition for this list.            
            if (!_partitionedConnectionPool.TryGetValue(_partitionName, out _actorClient))
            {
                // Optimistically create a new one, then only check for races after we take a lock further below.
                // Imperically, this performs faster than the obvious double-checked locking, though it can throw a lot more exceptions.
                _actorClient = FabricActorClient.CreateOrConnectToActor(hostEndpoint, ApplicationType, actorUri,
                                                        "ListActorReplica", connectThroughGateway, numReplicas);

                lock (_partitionedConnectionPool)
                {
                    FabricActorClient cachedClient = null;
                    if (!_partitionedConnectionPool.TryGetValue(_partitionName, out cachedClient))
                    {
                        _actorClient.IsPooled = true;
                        Thread.MemoryBarrier();
                        _partitionedConnectionPool.Add(_partitionName, _actorClient);
                    }
                    else {
                        _actorClient.Dispose();
                        _actorClient = cachedClient;
                    }
                }
            }
            
            // Deploy T's assembly to the server, if necessary.
            // @TODO: Look at types T depends on.  IE, is T a Tuple<UserType1, UserType2>?
            _actorClient.DeployAssemblyIfNeeded(typeof(T).Assembly);

            _actorClient.CallMethod("TryConstruct", ListNameOneElementArray);
        }
#endif

        // Use built-in partitioning support, which is somewhat incompatible with a naive approach to connection pooling.
        // Connection pooling would need to be done after we've resolved the right service partition, in FabricActorClient.
        public CloudList(String hostEndpoint, String listName, String partitionName, int numPartitions, 
            /*CancellationToken cancellationToken,*/ bool connectThroughGateway, int numReplicas = 2)
        {
            Contract.Requires(!String.IsNullOrEmpty(hostEndpoint));
            Contract.Requires(!String.IsNullOrEmpty(listName));
            Contract.Requires(!String.IsNullOrEmpty(partitionName));
            Contract.Requires(numPartitions >= 1, "Actors must have one or more partitions.");
            Contract.Requires(numReplicas >= 1, "Actors must have at least one instance of their state in the cloud.");

            Uri actorUri = new Uri(ApplicationType + "/" + listName);
            // With partitioned actors, the list name isn't particularly important.
            _listName = listName;
            ListNameOneElementArray = new Object[] { _listName };
            _partitionName = partitionName;

            // See if this collection exists.  If not, create it.
            _actorClient = ActorClient.CreateOrConnectToActor(hostEndpoint, ApplicationType, actorUri,
                                                              "ListActorReplica", connectThroughGateway, numReplicas, numPartitions, _partitionName);

            // Deploy T's assembly to the server, if necessary.
            // @TODO: Look at types T depends on.  IE, is T a Tuple<UserType1, UserType2>?
            // Update: We no longer need .NET type information on the actor.  It's all JSON on the actor.
            //_actorClient.DeployAssemblyIfNeeded(typeof(T).GetTypeInfo().Assembly);

            _actorClient.CallMethod("TryConstruct", ListNameOneElementArray);
            //_cancellationToken = cancellationToken;
        }


        #region Lifetime management

        ~CloudList()
        {
            // Ensure we close the ActorClient.
            Dispose(false);
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (_actorClient != null)
            {
                if (disposing)
                {
                    if (_deleteActor)
                    {
                        // @TODO: I have not heard that there is a way to delete one single partitioned data structure within a 
                        // partition, only the ability to delete the entire service instance.  Figure this out with our partners.
                        if (_partitionName == null)
                            _actorClient.DeleteActor();
                    }
                }

                // Note that we're calling Dispose() here on _actorClient, which may have theoretically already been 
                // finalized if disposing is false.  However, if someone creates a CloudList<T> and leaks it, we need 
                // some way to close the connection - remember we're doing an async read, so the continuation itself which isn't
                // even running is sitting on that socket and holding the ActorClient instance live.  Calling Dispose
                // on it will force the connection to close, and get rid of that continuation.
                _actorClient.Dispose();

                _actorClient = null;
            }
        }

        // Destroy the CloudList’s storage in the cloud
        public virtual void Destroy()
        {
            // @TODO: Revisit this once we talk about co-locating collections on a single actor.  Destroying the list != deleting the actor.
            _actorClient.CallMethod("Destroy", ListNameOneElementArray);
            String fabricAddress = _actorClient.ServerUri.Host + ":" + _actorClient.ServerUri.Port;
            _deleteActor = true;
            Dispose();
        }

        #endregion Lifetime management

        #region IAsyncList<T> implementation

        // @TODO: Not yet plumbed through to FabricActorClient yet.
        public TimeSpan Timeout
        {
            get;
            set;
        }

        public Task<T> GetItemAsync(int index)
        {
            return _actorClient.CallMethodAsync<T>("GetAt", new Object[] { _listName, index });
        }

        public Task SetItemAsync(int index, T value)
        {
            return _actorClient.CallMethodAsyncVoid("UpdateItemAt", new Object[] { _listName, index, value });
        }

        public Task<int> IndexOfAsync(T item)
        {
            return _actorClient.CallMethodAsync<int>("IndexOf", new Object[] { _listName, item });
        }

        public Task InsertAsync(int index, T item)
        {
            return _actorClient.CallMethodAsync<Object>("InsertAt", new Object[] { _listName, index, item });
        }

        public Task RemoveAtAsync(int index)
        {
            return _actorClient.CallMethodAsyncVoid("RemoveAt", new Object[] { _listName, index });
        }

        public Task<int> CountAsync
        {
            get
            {
                Task<int> task = _actorClient.CallMethodAsync<int>("Length", ListNameOneElementArray);
                return task;
            }
        }

        public Task<bool> IsReadOnlyAsync
        {
            get { return _actorClient.CallMethodAsync<bool>("IsReadOnly", ListNameOneElementArray); }
        }

        public Task AddAsync(T item)
        {
            return _actorClient.CallMethodAsync<int>("Push", new Object[] { _listName, "right", item });
        }

        public Task ClearAsync()
        {
            return _actorClient.CallMethodAsyncVoid("Clear", ListNameOneElementArray);
        }

        public Task<bool> ContainsAsync(T item)
        {
            Task<int> task = _actorClient.CallMethodAsync<int>("IndexOf", new Object[] { _listName, item });
            return task.ContinueWith((completedTask) => completedTask.Result >= 0);
        }

        public Task CopyToAsync(T[] array, int arrayIndex)
        {
            Task<List<Object>> task = _actorClient.CallMethodAsync<List<Object>>("Enumerate", new Object[] { _listName, 0, -1 });
            return task.ContinueWith((completedTask) =>
            {
                List<Object> snapshot = task.Result;
                for (int i = 0; i < snapshot.Count; i++)
                    array[arrayIndex + i] = (T)snapshot[i];
            });
        }

        public Task<bool> RemoveAsync(T item)
        {
            return _actorClient.CallMethodAsync<bool>("Remove", new Object[] { _listName, item });
        }

        public IDisposable Subscribe(IObserver<T> observer)
        {
            IObserver<ActorEvent> innerObserver = Observer.Create<ActorEvent>((actorEvent) =>
            {
                T item = actorEvent.GetTypedPayload<T>();
                observer.OnNext(item);
            });
            return _actorClient.Subscribe("*", innerObserver);
        }

        protected IDisposable SubscribeReal(String eventType, IObserver<ActorEvent> observer)
        {
            return _actorClient.Subscribe(eventType, observer);
        }

        #endregion IAsyncList<T> implementation

        #region Other List-like methods we should implement

        // LINQ has a ToArray method, but it will fail if the number of elements in the list increases while calling ToArray.
        // Our tests will call ToArray on a CloudList<T> while other threads are adding elements to it.  By doing the operation 
        // ourselves, we can avoid this race that breaks LINQ's implementation.
        public T[] ToArray()
        {
            return ToArrayAsync().Result;
        }

        public async Task<T[]> ToArrayAsync()
        {
            // Enumerate returns a List<Object> on the server side, but we can tell our JSON serializer to deserialize it
            // as a List<T> and let it figure out the type conversions.
            Task<List<T>> task = _actorClient.CallMethodAsync<List<T>>("Enumerate", new Object[] { _listName, 0, -1 });
            List<T> list = await task.ConfigureAwait(continueOnCapturedContext:false);
            return list.ToArray();
        }

        public Task SortAsync()
        {
            return _actorClient.CallMethodAsyncVoid("Sort", ListNameOneElementArray);
        }

        // Returns one array containing the two objects then each value within the array parameter.
        private static Object[] FlattenArray(Object obj1, Object obj2, T[] array)
        {
            Object[] parameters = new Object[array.Length + 2];
            parameters[0] = obj1;
            parameters[1] = obj2;
            Array.Copy(array, 0, parameters, 2, array.Length);
            return parameters;
        }

        private static int GetApproximateSerializedSize(Type t, T[] examples)
        {
            int serializedSizeInBytes = 100;  // Default size for reference types.  For JSON serialization w/ type information, ~100 is reasonable.
            T example = default(T);
            if (t.GetTypeInfo().IsClass)
            {
                // Find a non-null instance and try serializing it.
                foreach (T item in examples)
                {
                    if (item != null)
                    {
                        example = item;
                        break;
                    }
                }
            }

            if (example != null)
            {
                String serialized = SerializationHelper.Serialize(example);
                serializedSizeInBytes = serialized.Length;
            }
            return serializedSizeInBytes;
        }

        public Task AddAsync(IEnumerable<T> items)
        {
            Task<int> pushTask;
            T[] array = items.ToArray<T>();
            // We used to depend on WCF, which had a limitation of 64 KB that you could send over a socket in one call.
            // We've ripped out WCF now, so this restriction is no longer needed.
            // However, performance is significantly more even if we do chunking around ~1 MB in size, vs. using multiple
            // megabyte payloads.
            //
            // Curiously, we must do this based on the size of serialized T's.  What size is that?
            // We can use the current object size as a rough approximation, but if serialization requires huge headers...
            const int maxSize = 1 << 20;
            int serializedSizeInBytes = GetApproximateSerializedSize(typeof(T), array);
            if (serializedSizeInBytes * array.Length < maxSize)
            {
                Object[] parameters = FlattenArray(_listName, "right", array);
                pushTask = _actorClient.CallMethodAsync<int>("Push", parameters);
                return pushTask;
            }
            else
            {
                int chunkSize = maxSize / serializedSizeInBytes - 1000;  // Minus 1000 for random serialization overhead
                Task task = Task.Factory.StartNew(() =>
                {
                    T[] chunk;
                    if (chunkSize >= array.Length)
                        chunk = array;
                    else
                        chunk = new T[chunkSize];
                    Task<int> addChunk = null;
                    int numWholeChunks = array.Length / chunkSize;
                    for (int i = 0; i < numWholeChunks; i++)
                    {
                        Array.Copy(array, i * chunkSize, chunk, 0, chunkSize);
                        if (addChunk != null)
                            addChunk.Wait();
                        Object[] parameters = FlattenArray(_listName, "right", chunk);
                        addChunk = _actorClient.CallMethodAsync<int>("Push", parameters);
                    }
                    addChunk.Wait();

                    int numLeft = array.Length - numWholeChunks * chunkSize;
                    if (numLeft > 0)
                    {
                        chunk = new T[numLeft];
                        Array.Copy(array, numWholeChunks * chunkSize, chunk, 0, numLeft);
                        Object[] parameters = FlattenArray(_listName, "right", chunk);
                        addChunk = _actorClient.CallMethodAsync<int>("Push", parameters);
                        addChunk.Wait();
                    }
                });
                return task;
            }
        }

        public Task AddOnceAsync(IEnumerable<T> items)
        {
            Object[] parameters = new Object[1 + items.Count()];
            parameters[0] = _listName;
            int i = 1;
            foreach (T item in items)
                parameters[i++] = item;
            return _actorClient.CallMethodAsync<int>("AddOnce", parameters);
        }

        public Task RemoveRangeAsync(int index, int count)
        {
            return _actorClient.CallMethodAsyncVoid("RemoveRange", new Object[] { _listName, index, count });
        }

        #endregion Other List-like methods we should implement

        #region IList<T> implementation

        public int IndexOf(T item)
        {
            return _actorClient.CallMethod<int>("IndexOf", new Object[] { _listName, item });
        }

        public void Insert(int index, T item)
        {
            _actorClient.CallMethod("InsertAt", new Object[] { _listName, index, item });
        }

        public void RemoveAt(int index)
        {
            _actorClient.CallMethod("RemoveAt", new Object[] { _listName, index });
        }

        public T this[int index]
        {
            get
            {
                return _actorClient.CallMethod<T>("GetAt", new Object[] { _listName, index });
            }
            set
            {
                _actorClient.CallMethod("UpdateItemAt", new Object[] { _listName, index, value });
            }
        }

        public void Add(T item)
        {
            _actorClient.CallMethod("Push", new Object[] { _listName, "right", item });
        }

        public void Clear()
        {
            _actorClient.CallMethod("Clear", ListNameOneElementArray);
        }

        public bool Contains(T item)
        {
            int index = _actorClient.CallMethod<int>("IndexOf", new Object[] { _listName, item });
            return index >= 0;
        }

        public void CopyTo(T[] array, int arrayIndex)
        {
            List<T> snapshot = _actorClient.CallMethod<List<T>>("Enumerate", new Object[] { _listName, 0, -1 });
            for (int i = 0; i < snapshot.Count; i++)
                array[arrayIndex + i] = (T)snapshot[i];
        }

        public virtual int Count
        {
            get { return _actorClient.CallMethod<int>("Length", ListNameOneElementArray); }
        }

        public bool IsReadOnly
        {
            get { return _actorClient.CallMethod<bool>("IsReadOnly", ListNameOneElementArray); }
        }

        public bool Remove(T item)
        {
            int numRemoved = _actorClient.CallMethod<int>("Remove", new Object[] { _listName, 1, item });
            return numRemoved != 0;
        }

        public virtual IEnumerator<T> GetEnumerator()
        {
            List<T> contents = _actorClient.CallMethod<List<T>>("Enumerate", new Object[] { _listName, 0, -1 });
            return contents.GetEnumerator();
            //T[] array = new T[contents.Count];
            //for (int i = 0; i < contents.Count; i++)
            //    array[i] = (T)contents[i];
            //return ((IEnumerable<T>)array).GetEnumerator();
        }

        System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
        {
            return GetEnumerator();
        }

        #endregion IList<T> implementation

    }

    public class ObservableCloudList<T> : CloudList<T>, IReadOnlyObservableList<T>, INotifyCollectionChanged
    {
        private SynchronizationContext _uiThreadSyncContext;
        private IDisposable _subscription;
        private int _threadCallingGetEnumeratorMustReturnEmpty;  // Used to be a Thread, but PLib provides only Env.CurrentManagedThreadId.
        private int _cachedCount;   // data binding keeps calling Count, all the time.

        public ObservableCloudList(String server, int port, String listName, SynchronizationContext uiThreadSyncContext, bool useGateway = false)
            : this(server + ':' + port.ToString(), listName, uiThreadSyncContext, useGateway/*, CancellationToken.None*/)
        {
            Contract.Requires(!String.IsNullOrEmpty(listName));
        }

        /*
        public ObservableCloudList(String server, int port, String listName, SynchronizationContext uiThreadSyncContext, CancellationToken cancellationToken)
            : this(server + ':' + port.ToString(), listName, uiThreadSyncContext, cancellationToken)
        {
        }
        */

        public ObservableCloudList(String hostEndpoint, String listName, SynchronizationContext uiThreadSyncContext, /*CancellationToken cancellationToken,*/ bool useGateway = false, int numReplicas = 2)
            : base(hostEndpoint, listName, /*cancellationToken,*/ useGateway, numReplicas)
        {
            Contract.Requires(numReplicas >= 1, "Actors must have at least one instance of their state in the cloud.");

            if (uiThreadSyncContext == null)
                throw new ArgumentNullException("uiThreadSyncContext");

            _uiThreadSyncContext = uiThreadSyncContext;
            _subscription = SubscribeReal("*", new NotificationObserver(this));
            _cachedCount = -1;
        }

        public ObservableCloudList(String hostEndpoint, String listName, String partitionName, int numPartitions,
            SynchronizationContext uiThreadSyncContext, /*CancellationToken cancellationToken,*/ bool useGateway = false, int numReplicas = 2)
            : base(hostEndpoint, listName, partitionName, numPartitions, /*cancellationToken,*/ useGateway, numReplicas)
        {
            Contract.Requires(numPartitions >= 1, "Actors must have one or more partitions.");
            Contract.Requires(numReplicas >= 1, "Actors must have at least one instance of their state in the cloud.");

            if (uiThreadSyncContext == null)
                throw new ArgumentNullException("uiThreadSyncContext");

            _uiThreadSyncContext = uiThreadSyncContext;
            _subscription = SubscribeReal("*", new NotificationObserver(this));
            _cachedCount = -1;
        }

        protected override void Dispose(bool disposing)
        {
            if (disposing)
            {
                if (_subscription != null)
                    _subscription.Dispose();
                _subscription = null;
            }
            base.Dispose(disposing);
        }

        public override void Destroy()
        {
            if (_subscription != null)
            {
                _subscription.Dispose();
                _subscription = null;
            }
            base.Destroy();
        }

        public event NotifyCollectionChangedEventHandler CollectionChanged;

        public override IEnumerator<T> GetEnumerator()
        {
            // This is implemented this way to support Clear.  WPF's support for INotifyCollectionChanged events
            // doesn't support Clear, so our Clear message from the server is implemented as a reset of the collection,
            // which requires calling GetEnumerator().  There's a subtle race where we publish a Clear message to a client,
            // but it then calls GetEnumerator later, and either another client accessing the list or this client could have
            // already added more items.  This race leads to seeing the list contents duplicated, like A B C A B C.
            // So we need to explicitly lie from GetEnumerator sometimes and say the list is empty - it was empty at one point in time!
            if (_threadCallingGetEnumeratorMustReturnEmpty == Environment.CurrentManagedThreadId)
            {
                _threadCallingGetEnumeratorMustReturnEmpty = -1;
                return Enumerable.Empty<T>().GetEnumerator();
            }
            return base.GetEnumerator();
        }

        public override int Count {
            get
            {
                if (_cachedCount == -1)
                    _cachedCount = base.Count;
                return _cachedCount;
            }
        }

        private class NotificationObserver : IObserver<ActorEvent>
        {
            private ObservableCloudList<T> _observableList;

            internal NotificationObserver(ObservableCloudList<T> list)
            {
                _observableList = list;
            }

            public void OnCompleted()
            {
                _observableList = null;
            }

            public void OnError(Exception error)
            {
                throw new NotImplementedException("NotificationObserver::OnError NYI");
            }

            public void OnNext(ActorEvent value)
            {
                _observableList._cachedCount = -1;  // We could calculate the count from the arguments, but we get a big bang 
                // for the buck by very limited caching, because the Windows Phone data binding keeps calling Count.

                NotifyCollectionChangedEventHandler localHandler = _observableList.CollectionChanged;
                if (localHandler != null)
                {
                    NotifyCollectionChangedEventArgs args = ToEventArgs(value);

                    // Notifications must take place on the UI thread, at least in WPF.  WPF doesn't automatically
                    // capture the synchronization context for us.  So, we'll ask the user to pass in the synchronization context.
                    _observableList._uiThreadSyncContext.Post(new SendOrPostCallback((obj) =>
                    {
                        // This is a horrible hack for Clear notifications, where using Reset to implement them works poorly.
                        // We can hit a race with GetEnumerator() when multiple clients can be adding items to the collection.
                        // We avoid this by making GetEnumerator() return an empty IEnumerable<T>, but only precisely once on this thread.
                        if (args.Action == NotifyCollectionChangedAction.Reset)
                            this._observableList._threadCallingGetEnumeratorMustReturnEmpty = Environment.CurrentManagedThreadId;
                        localHandler(this, args);
                    }), null);
                }
            }

            private static NotifyCollectionChangedEventArgs ToEventArgs(ActorEvent actorEvent)
            {
                String action = actorEvent.EventType;
                NotifyCollectionChangedEventArgs args;
                T item;
                int index;
                switch (action)
                {
                    case "Collection.ElementAdded":
                        item = actorEvent.GetTypedPayload<T>();
                        index = actorEvent.GetTypedPayloadQualifier<int>(0); 
                        args = new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Add, item, index);
                        break;

                    case "Collection.ElementRemoved":
                        item = actorEvent.GetTypedPayload<T>();
                        index = actorEvent.GetTypedPayloadQualifier<int>(0); 
                        args = new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Remove, item, index);
                        break;

                    case "Collection.Cleared":
                        args = new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Reset);
                        break;

                    /*  We have no analogy for NotifyCollectionChangedAction.Move.
                case NotifyCollectionChangedAction.Move:
                    args = new NotifyCollectionChangedEventArgs(_action, _newItems, _newStartingIndex, _oldStartingIndex);
                    break;
                    */

                    case "Collection.ElementReplaced":
                        item = actorEvent.GetTypedPayload<T>();
                        T oldItem = actorEvent.GetTypedPayloadQualifier<T>(0);
                        index = actorEvent.GetTypedPayloadQualifier<int>(1); 
                        args = new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Replace, new Object[] { item }, new Object[] { oldItem }, index);
                        break;

                    default:
                        String message = String.Format("Unknown publish event from our ListActor.  Published event type: {0}", action);
                        Contract.Assume(false, message);
                        throw new InvalidOperationException(message);
                }
                return args;
            }
        }
    }
}
